package org.event.core.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;

/**
 * topic工具类
 *
 * @author zhangjm
 * @date 2023年11月29日 11:28
 */
@Slf4j
public class KafkaTopicUtil {


    /**
     * 功能描述:创建topic，并返回创建结果
     *
     * @param: topicName
     * @return: boolean
     * @auther: lyp
     * @date: 2021/11/12 16:06
     */
    public static boolean createTopics(String bootstrapServers, String topicName, int partitions, short replication) {
        boolean res = false;
        try {
            Properties properties = new Properties();
            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            properties.put("sasl.jaas.config",
                    "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";");
            AdminClient adminClient = KafkaAdminClient.create(properties);
            NewTopic newTopic = new NewTopic(topicName, partitions, replication);
            adminClient.createTopics(Arrays.asList(newTopic));
            log.info("创建Topic:" + topicName + "成功!");
            res = true;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("创建异常！");
        }
        return res;
    }

    /**
     * 功能描述:获取当前kafka所存在的topic列表
     *
     * @return: set
     * @auther: lyp
     * @date: 2021/11/12 16:07
     */
    public static Set<String> getTopics(String bootstrapServers) {
        Set<String> nameSet = new HashSet<>();
        try {
            Properties properties = new Properties();
            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            AdminClient adminClient = KafkaAdminClient.create(properties);
            ListTopicsResult listTopicsResult = adminClient.listTopics();
            KafkaFuture<Set<String>> names = listTopicsResult.names();
            nameSet = names.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return nameSet;
    }

}
